/*
 * hashring.c
 *
 *  Created on: 2012-4-1
 *      Author: zhanggx
 */

#include "gx_hashring_purenodeimpl.h"

#define ESTIMATE_SAME_VALUE_NODE_COUNT 3

int
gx_pure_copy_to_ring(const gx_hashring_t* ring, gx_hashring_t* another_ring){
	another_ring->copy_to_ring 			= ring->copy_to_ring;
	another_ring->init_ring    			= ring->init_ring;
	another_ring->select_node_for_key 	= ring->select_node_for_key;
	another_ring->destroy_ring 			= ring->destroy_ring;
	another_ring->add_node  			= ring->add_node;
	another_ring->delete_node 			= ring->delete_node;

	gx_pure_init_ring(another_ring);
	return 0;
}

int gx_pure_init_ring(gx_hashring_t* ring){
	int ret;

	ring->node_tree = (gx_rbtree_t*)gx_malloc(sizeof(gx_rbtree_t));

	ret = gx_rbtree_init(ring->node_tree);
	ring->node_count  = 0;
	ring->vnode_count = 0;

	if(ret != 0)
		return ret;
	return 0;
}
int gx_pure_add_node(gx_hashring_t* ring, gx_string_t key, void* user_data) {
	assert(ring);

	gx_hash_node_t   *new_hash_node;
	int ret;

	new_hash_node = (gx_hash_node_t *)gx_malloc(sizeof(gx_hash_node_t));
	if(!new_hash_node){
		GX_ERROR_LOG("malloc error");
		return -1;
	}

	gx_set_string(new_hash_node->key, key.buf);
	new_hash_node->usr_data = user_data;

	gx_rbtree_key_t hash_value = gx_globals.g_hash->hash(key.buf, key.len);

	ret = gx_rbtree_insert(ring->node_tree, hash_value, new_hash_node);
	if(ret != 0){
		GX_ERROR_LOG("insert to rbtree error");
		return ret;
	}

	GX_NOTICE_LOG("new node value %s has inserted to ring", key.buf);
	ring->node_count++;
	return 0;
}


int
gx_pure_delete_node(gx_hashring_t* ring, gx_string_t key) {
	assert(ring);
	gx_array_t node_list;
	int ret, i;

	GX_NOTICE_LOG("hashring delete %s", key.buf);

	ret = gx_array_init(&node_list, ESTIMATE_SAME_VALUE_NODE_COUNT,
			sizeof(gx_rbtree_node_t*));
	if(ret != 0){
		GX_WARNING_LOG("init node list error");
		return ret;
	}

	gx_rbtree_key_t hash_value;
	hash_value = gx_globals.g_hash->hash(key.buf, key.len);

	ret = gx_rbtree_search(ring->node_tree, hash_value, &node_list);

	if(node_list.nelts == 0) {
		GX_NOTICE_LOG("not founded node");
		return -1;
	}

	//delete node with same key of node
	for(i = 0; i < node_list.nelts; i++){
		gx_rbtree_node_t** node = gx_array_get(&node_list, i);
		gx_hash_node_t* hash_node = (gx_hash_node_t*)((*node)->usr_data);
		if(gx_is_same_string(hash_node->key, key)){
			ret = gx_rbtree_delete(ring->node_tree, *node);
			if(0 != ret){
				GX_WARNING_LOG("can't delete node from rbtree");
				return -1;
			}
			ring->node_count--;
			break;
		}
	}

	return 0;
}
#define ESTIMATE_SAME_NODE_COUNT 3
int gx_pure_select_node(gx_hashring_t* ring, gx_hash_node_t* node, gx_string_t key){
	gx_uint32_t value;

	value = gx_globals.g_hash->hash(key.buf, key.len);

	gx_array_t node_list;
	gx_array_init(&node_list, ESTIMATE_SAME_NODE_COUNT, sizeof(gx_rbtree_node_t*));

	int ret = gx_rbtree_search_nearest(ring->node_tree, value, &node_list);
	if(ret != 0){
		GX_ERROR_LOG("get nearest node error");
		return ret;
	}

	gx_rbtree_node_t** tree_node;
	if(node_list.nelts <= 0){
		GX_WARNING_LOG("no node could be found");
		return -1;
	}
	else if (node_list.nelts == 1){
		tree_node = (gx_rbtree_node_t **)gx_array_get(&node_list, 0);
	}
	else{
		//TODO how to handle multiple hash collision
		tree_node = (gx_rbtree_node_t **)gx_array_get(&node_list, 0);
	}

	*node = *(gx_hash_node_t *)((*tree_node)->usr_data);
	GX_NOTICE_LOG("select node %s serve for key %s", node->key.buf, key.buf);
	return 0;
}
